package com.leilei.time.slide;

import com.google.common.collect.Lists;
import com.leilei.Location;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.UUID;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 16:52
 */
public class SlidingWindowFunction extends RichWindowFunction<Location,List<Location>, Integer,TimeWindow> {
    public SlidingWindowFunction() {}
    String uuid;
    /**
     * @param window 当前窗口
     * @param locations 入站数据集合
     * @param out 出站数据收集器 泛型为出站数据类型
     */
    @Override
    public void apply(Integer integer, TimeWindow window, Iterable<Location> locations, Collector<List<Location>> out) {
        out.collect(Lists.newArrayList(locations));
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        uuid = UUID.randomUUID().toString();
        System.out.println(uuid + "窗口打开了");
    }

    @Override
    public void close() throws Exception {
        System.out.println(uuid + "窗口关闭了");
    }
}
